package com.qqt.csr.im.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.im.assembler.AccountAssembler;
import com.qqt.csr.im.entity.Account;
import com.qqt.csr.im.entity.Connection;
import com.qqt.csr.im.enums.OnlineEnum;
import com.qqt.csr.im.mapper.AccountMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.BatchResult;
import org.redisson.api.RBatch;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2024/10/16 9:59
 * @Description:
 **/
@Slf4j
@Service
public class AccountService extends ServiceImpl<AccountMapper, Account> {
    /**
     * 在线本地缓存连接，key:accountId
     */
    private Cache<Long, Account> ACCOUNT_ID_CACHE = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterAccess(30, TimeUnit.SECONDS)
            .build();
    /**
     * 在线本地缓存连接，key:accountUserId
     */
    private Cache<String, Account> ACCOUNT_USER_ID_CACHE = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterAccess(30, TimeUnit.SECONDS)
            .build();
    @Autowired
    private ConnectionService connectionService;
    @Autowired
    private RedissonLockUtil redissonLockUtil;
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 保存或更新
     *
     * @param accountUserId 用户标识
     * @return {@link Boolean} 保存或更新是否成功
     */
    public Account save(String accountUserId, String name, String headImg, String ext) {
        String lockKey = String.format(CacheKeyUtil.IM.CREATE_ACCOUNT_LOCK, accountUserId);
        return redissonLockUtil.distributedLock(() -> {
            Account account = getByAccountUserId(accountUserId);
            if (account == null) {
                account = AccountAssembler.convert(accountUserId, name, headImg);
                account.setExt(ext);
                save(account);
            } else {
                Account updateAccount = Account.builder()
                        .id(account.getId())
                        .accountName(name)
                        .headImg(headImg)
                        .ext(ext)
                        .build();
                updateById(updateAccount);
                clearCache(account);
            }
            return account;
        }, lockKey, 0, 30, TimeUnit.SECONDS);
    }


    /**
     * 更新用户上线状态
     *
     * @param accountId
     */
    public void updateOnline(Long accountId, String accountUserId) {
        this.lambdaUpdate().set(Account::getType, OnlineEnum.ONLINE.getCode()).set(Account::getUpdateTime, new Date())
                .eq(Account::getId, accountId)
                .eq(Account::getType, OnlineEnum.DOWNLINE.getCode())
                .update();
        clearCache(Account.builder().id(accountId).accountUserId(accountUserId).build());
    }

    /**
     * 更新用户下线状态
     *
     * @param accountId
     */
    public void updateOffline(Long accountId, String accountUserId) {
        this.lambdaUpdate().set(Account::getType, OnlineEnum.DOWNLINE.getCode())
                .set(Account::getUpdateTime, new Date())
                .eq(Account::getId, accountId)
                .eq(Account::getType, OnlineEnum.ONLINE.getCode())
                .update();
        clearCache(Account.builder().id(accountId).accountUserId(accountUserId).build());
    }

    /**
     * 更新用户下线状态
     *
     * @param accountIdList
     */
    public void updateOffline(List<Long> accountIdList) {
        if (CollectionUtils.isEmpty(accountIdList)) {
            return;
        }
        this.lambdaUpdate().set(Account::getType, OnlineEnum.DOWNLINE.getCode())
                .set(Account::getUpdateTime, new Date())
                .in(Account::getId, accountIdList)
                .eq(Account::getType, OnlineEnum.ONLINE.getCode())
                .update();

        clearCache(this.lambdaQuery().select(Account::getId, Account::getAccountUserId).in(Account::getId, accountIdList).list());
    }

    /**
     * 获取账户信息
     *
     * @param accountUserId 用户标识
     * @return {@link Account} 账户信息
     */
    public Account getByAccountUserId(String accountUserId) {
        if (StringUtils.isBlank(accountUserId)) {
            return null;
        }
        Account account = ACCOUNT_USER_ID_CACHE.getIfPresent(accountUserId);
        if (account != null) {
            return account;
        }

        String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, accountUserId);
        account = Optional.ofNullable(redissonClient.getBucket(redisKey).get()).map(Object::toString)
                .map(o -> JsonUtil.toObject(o, Account.class)).orElse(null);
        if (account != null) {
            addLocalCache(account);
            return account;
        }

        account = this.lambdaQuery().select().eq(Account::getAccountUserId, accountUserId).one();
        if (account != null) {
            addCache(account);
        }
        return account;
    }

    /**
     * 获取账户信息
     *
     * @param accountId 用户标识
     * @return {@link Account} 账户信息
     */
    public Account getByAccountId(Long accountId) {
        if (accountId == null) {
            return null;
        }
        Account account = ACCOUNT_ID_CACHE.getIfPresent(accountId);
        if (account != null) {
            return account;
        }

        String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, accountId);
        account = Optional.ofNullable(redissonClient.getBucket(redisKey).get()).map(Object::toString)
                .map(o -> JsonUtil.toObject(o, Account.class)).orElse(null);
        if (account != null) {
            addLocalCache(account);
            return account;
        }

        account = this.lambdaQuery().select().eq(Account::getId, accountId).one();
        if (account != null) {
            addCache(account);
        }
        return account;
    }

    /**
     * 根据userIds获取账户集合
     *
     * @param accountUserIdList {@link List<String>} 用户ids
     * @return {@link List<Account>} 账户列表
     */
    public Map<String, Account> getMapByUserIds(List<String> accountUserIdList) {
        if (CollectionUtils.isEmpty(accountUserIdList)) {
            return Maps.newHashMap();
        }
        Map<String, Account> accountMap = Maps.newHashMap();

        // 本地缓存获取
        List<String> needQueryRedisAccountUserIdList = new ArrayList<>();
        accountUserIdList.forEach(accountUserId -> {
            Account account = ACCOUNT_USER_ID_CACHE.getIfPresent(accountUserId);
            if (account != null) {
                accountMap.put(accountUserId, account);
            } else {
                needQueryRedisAccountUserIdList.add(accountUserId);
            }
        });
        if (CollectionUtils.isEmpty(needQueryRedisAccountUserIdList)) {
            return accountMap;
        }

        // redis缓存获取
        RBatch redisBatch = redissonClient.createBatch();
        for (String accountUserId : needQueryRedisAccountUserIdList) {
            String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, accountUserId);
            redisBatch.getBucket(redisKey).getAsync();
        }
        BatchResult<?> batchResult = redisBatch.execute();
        List<Account> accountJsonList = Optional.ofNullable(batchResult.getResponses()).orElseGet(Lists::newArrayList)
                .stream().filter(Objects::nonNull).map(Object::toString).map(json -> JsonUtil.toObject(json, Account.class)).collect(Collectors.toList());
        accountJsonList.forEach(account -> {
            accountMap.put(account.getAccountUserId(), account);
            addLocalCache(account);
        });

        List<String> needQueryDbAccountUserIdList = needQueryRedisAccountUserIdList.stream()
                .filter(accountUserId -> !accountMap.containsKey(accountUserId)).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(needQueryDbAccountUserIdList)) {
            return accountMap;
        }

        List<Account> dbList = this.lambdaQuery().select().in(Account::getAccountUserId, needQueryDbAccountUserIdList).list();
        dbList.forEach(account -> accountMap.put(account.getAccountUserId(), account));
        addCache(dbList);
        return accountMap;
    }

    /**
     * 根据主键获取账户信息
     *
     * @param accountIdList
     * @return
     */
    public Map<Long, Account> getMapByIds(List<Long> accountIdList) {
        if (CollectionUtils.isEmpty(accountIdList)) {
            return Maps.newHashMap();
        }
        Map<Long, Account> accountMap = Maps.newHashMap();

        // 本地缓存获取
        List<Long> needQueryRedisAccountIdList = new ArrayList<>();
        accountIdList.forEach(accountId -> {
            Account account = ACCOUNT_ID_CACHE.getIfPresent(accountId);
            if (account != null) {
                accountMap.put(accountId, account);
            } else {
                needQueryRedisAccountIdList.add(accountId);
            }
        });
        if (CollectionUtils.isEmpty(needQueryRedisAccountIdList)) {
            return accountMap;
        }

        // redis缓存获取
        RBatch redisBatch = redissonClient.createBatch();
        for (Long accountId : needQueryRedisAccountIdList) {
            String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, accountId);
            redisBatch.getBucket(redisKey).getAsync();
        }
        BatchResult<?> batchResult = redisBatch.execute();
        List<Account> accountJsonList = Optional.ofNullable(batchResult.getResponses()).orElseGet(Lists::newArrayList)
                .stream().filter(Objects::nonNull).map(Object::toString).map(json -> JsonUtil.toObject(json, Account.class)).collect(Collectors.toList());
        accountJsonList.forEach(account -> {
            accountMap.put(account.getId(), account);
            addLocalCache(account);
        });

        List<Long> needQueryDbAccountIdList = needQueryRedisAccountIdList.stream()
                .filter(accountId -> !accountMap.containsKey(accountId)).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(needQueryDbAccountIdList)) {
            return accountMap;
        }

        // 数据库获取
        List<Account> dbList = this.lambdaQuery().select().in(Account::getId, needQueryDbAccountIdList).list();
        dbList.forEach(account -> accountMap.put(account.getId(), account));
        addCache(dbList);
        return accountMap;
    }

    private void addCache(Account account) {
        if (account == null) {
            return;
        }
        addLocalCache(account);

        String accountJson = JsonUtil.toJson(account);
        redissonClient.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, account.getId()))
                .setAsync(accountJson, CacheKeyUtil.EXPIRE_TIME_1H, TimeUnit.MILLISECONDS);
        redissonClient.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, account.getAccountUserId()))
                .setAsync(accountJson, CacheKeyUtil.EXPIRE_TIME_1H, TimeUnit.MILLISECONDS);
    }

    private void addCache(List<Account> accountList) {
        if (CollectionUtils.isEmpty(accountList)) {
            return;
        }

        RBatch redisBatch = redissonClient.createBatch();
        for (Account account : accountList) {
            addLocalCache(account);

            String accountJson = JsonUtil.toJson(account);
            redisBatch.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, account.getId()))
                    .setAsync(accountJson, CacheKeyUtil.EXPIRE_TIME_1H, TimeUnit.MILLISECONDS);
            redisBatch.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, account.getAccountUserId()))
                    .setAsync(accountJson, CacheKeyUtil.EXPIRE_TIME_1H, TimeUnit.MILLISECONDS);
        }
        redisBatch.execute();
    }

    private void addLocalCache(Account account) {
        ACCOUNT_ID_CACHE.put(account.getId(), account);
        ACCOUNT_USER_ID_CACHE.put(account.getAccountUserId(), account);
    }

    private void clearCache(Account account) {
        if (account == null) {
            return;
        }
        ACCOUNT_ID_CACHE.invalidate(account.getId());
        ACCOUNT_USER_ID_CACHE.invalidate(account.getAccountUserId());

        redissonClient.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, account.getId())).deleteAsync();
        redissonClient.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, account.getAccountUserId())).deleteAsync();
    }

    private void clearCache(List<Account> accountList) {
        if (CollectionUtils.isEmpty(accountList)) {
            return;
        }
        RBatch redisBatch = redissonClient.createBatch();
        accountList.forEach(account -> {
            ACCOUNT_ID_CACHE.invalidate(account.getId());
            ACCOUNT_USER_ID_CACHE.invalidate(account.getAccountUserId());

            redisBatch.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ID_KEY, account.getId())).deleteAsync();
            redisBatch.getBucket(String.format(CacheKeyUtil.IM.ACCOUNT_INFO_BY_ACCOUNT_USER_ID_KEY, account.getAccountUserId())).deleteAsync();
        });
        redisBatch.execute();
    }


}
